package cn.gwm.flink.streaming.strategy.cloudpredict;

import cn.gwm.flink.streaming.constant.ChargePredictFields;
import cn.gwm.flink.streaming.strategy.vehiclepredict.DwsStrConstant;
import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.java.typeutils.MapTypeInfo;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.streaming.api.functions.co.KeyedBroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.math.BigDecimal;
import java.util.Map;
import java.util.Set;
import java.util.StringJoiner;

/**
 * @author : GW00280745
 * @date : 2023/1/15 15:34
 * @description :
 */
@Slf4j
public class CloudPredictProcessFunction extends KeyedBroadcastProcessFunction<Object, JSONObject, JSONObject, JSONObject> {
    @Override
    public void processElement(JSONObject line, KeyedBroadcastProcessFunction<Object, JSONObject, JSONObject, JSONObject>.ReadOnlyContext readOnlyContext, Collector<JSONObject> out) throws Exception {
        try {
            process(line);
            out.collect(line);
        }catch (Exception e){
            //当前记录的异常不能影响下一条记录的处理进度
            e.printStackTrace();
        }
    }

    @Override
    public void processBroadcastElement(JSONObject confParam, KeyedBroadcastProcessFunction<Object, JSONObject, JSONObject, JSONObject>.Context ctx, Collector<JSONObject> collector) throws Exception {
        //广播变量
        ctx.getBroadcastState(confInfoMsd).put(null,confParam);
    }


    private void process(JSONObject line){
        //插枪后第一个点要进行预测
        firstMatch.process(line,initState,latestEstimateState);
        //P1 当进入电流充电状态一分钟后，电流相对稳定的，启动本次计算
        boolean isOneMin = oneMinute(line);
        boolean end = isEnd(line);
        if (!isOneMin || end){
            //不足1min或者大于98都不进行计算
            return;
        }
        //P2 识别匹配与否
        boolean neededMatch = isNeededMatch(line,0.99999f);
//        boolean neededMatch = isNeededMatch(line,0.49999f);
        if(!neededMatch){
            //不需要匹配 数据从state中读取
            state2Line(line);
            //记录当前切片的状态 进入计算阶段并且不是间隔点的数据
            lastOneSliceInfo(line);
            return;
        }
        long startTime = System.currentTimeMillis();
        //P2-1 需要匹配，首先需要 识别初始和更新
        boolean updatePhase = isUpdatePhase(line);
        //匹配
        cloudMatch.process(line,updatePhase,initState,latestEstimateState,timeOfLatest5State,startSocState);
        //P2-2 其次记录最近5个间隔点的工况
//        infoOfLatest5Point(line);
        infoOfLatest5PointV2(line,DwsStrConstant.ins().socInt);
//        infoOfLatest5PointV2(line,10);
        long endTime = System.currentTimeMillis();
        System.out.println(String.format("CloudPredictProcessFunction cost time ==> %s ms ",endTime - startTime));
    }

    private void infoOfLatest5PointV2(JSONObject line,int num) {
        try {
            String chargeId = chargeIdName(line);
            String tidVal = line.getString(ChargePredictFields.ins().srcFieldTid);
            String currStr = line.getString(ChargePredictFields.ins().srcFieldBmsBatterCurrRm);
            //更新点的状态信息
            String sliceInfo = lastOneSliceState.get(chargeId);
            if (StringUtils.isBlank(sliceInfo)){
                String concat = tidVal.concat(ChargePredictFields.ins().colon).concat(currStr);
                timeOfLatest5State.put(chargeId,concat);
                lastOneSliceState.put(chargeId,new JSONObject().toJSONString());
                return;
            }
            // (tid,curr) 间隔点的电流
            JSONObject jsonObject = JSONObject.parseObject(sliceInfo);
            float avgCurr = avgCurr(jsonObject);
            tidVal = tidVal.concat(ChargePredictFields.ins().colon).concat(String.valueOf(avgCurr));
            //清空切片信息
            lastOneSliceState.put(chargeId,new JSONObject().toJSONString());
            //2. 近5个点的信息
            String s = timeOfLatest5State.get(chargeId);
            //判断有几个点,不能大于5：判断现在有几个，等于5 和小于5
            String[] split = s.split(ChargePredictFields.ins().char3);
            if (split.length< num){
                //小于5 拼接
                String concat = s.concat(ChargePredictFields.ins().char3).concat(tidVal);
                timeOfLatest5State.put(chargeId,concat);
                return;
            }
            //等于5 说明需要去掉开头的
            StringJoiner sj = new StringJoiner(ChargePredictFields.ins().char3);
            for (int i = 1; i <split.length ; i++) {
                String item = split[i];
                sj = sj.add(item);
            }
            sj.add(tidVal);
            timeOfLatest5State.put(chargeId,sj.toString());
        }catch (Exception e){
            e.printStackTrace();
        }
    }

    private float avgCurr(JSONObject jsonObject) {
        Set<Map.Entry<String, Object>> entrySet = jsonObject.entrySet();
        float total = 0;
        for (Map.Entry<String, Object> entry : entrySet) {
            String key = entry.getKey();
            String value = entry.getValue().toString();
            float v = Float.parseFloat(value);
            total += v;
        }
        if (entrySet.size()==0){
            return 0;
        }
        float v = new BigDecimal(total / entrySet.size()).setScale(1, BigDecimal.ROUND_HALF_UP).floatValue();
        return v;
    }

    /**
     * 不是间隔点的数据作为下个间隔点的状态
     * @param line
     */
    private void lastOneSliceInfo(JSONObject line) {
        try {
            Float bmsSoc = line.getFloat(ChargePredictFields.ins().srcFieldBmsSoc);
            Float curr = line.getFloat(ChargePredictFields.ins().srcFieldBmsBatterCurrRm);
            String vin = line.getString(ChargePredictFields.ins().srcFieldVin);
            String tid = line.getString(ChargePredictFields.ins().srcFieldTid);
            String startChargeTime = line.getString(ChargePredictFields.ins().srcFieldStartChargeTime);
            String chargeId = chargeIdName(line);
            String s = lastOneSliceState.get(chargeId);
            JSONObject jsonObject = null;
            if (StringUtils.isBlank(s)){
                jsonObject = new JSONObject();
            }
            jsonObject = JSONObject.parseObject(s);
            jsonObject.put(tid,curr.toString());
            lastOneSliceState.put(chargeId,jsonObject.toJSONString());
        }catch (Exception e){
            e.printStackTrace();
        }
    }

    private String chargeIdName(JSONObject line){
        String vin = line.getString(ChargePredictFields.ins().srcFieldVin);
        String startChargeTime = line.getString(ChargePredictFields.ins().srcFieldStartChargeTime);
        String chargeId = vin.concat(startChargeTime);
        return chargeId;
    }
    /**
     *  是否是结束阶段
     * @param line
     * @return
     */
    private boolean isEnd(JSONObject line) {
        Float bmsSoc = line.getFloat(ChargePredictFields.ins().srcFieldBmsSoc);
        boolean b = bmsSoc > 101;
        if (!b){
            return false;
        }
        //大于98 不进行计算 沿用之前的计算结果
        try {

            String vin = line.getString(ChargePredictFields.ins().srcFieldVin);
            String startChargeTime = line.getString(ChargePredictFields.ins().srcFieldStartChargeTime);
            String chargeId = vin.concat(startChargeTime);
            String keyMult = ChargePredictFields.ins().stdFieldPredictTimeV2;
            String stateKeyMult = chargeId.concat(keyMult);
            String predict = latestEstimateState.get(stateKeyMult);
            if (StringUtils.isBlank(predict)){
                System.out.println(String.format("大于97但=空,%s",line.toJSONString()));
                return true;
            }
            //1
            String[] split = predict.split(ChargePredictFields.ins().char3);
            String estimateTime = split[0];
            line.put(ChargePredictFields.ins().stdFieldEstimateTime, estimateTime);
            //2
            line.put(keyMult, predict);
            //3
            String predictTime = cloudMatch.predictTime(line, estimateTime);
            String keyPredict = ChargePredictFields.ins().stdFieldPredictTime;
            line.put(keyPredict, predictTime);
        }catch (Exception e){
            e.printStackTrace();
        }
        return true;
    }

    /**
     *  记录最近5个点的信息
     * @param line
     */
    private void infoOfLatest5Point(JSONObject line) {
        try {
            //当前工况信息
            String vin = line.getString(ChargePredictFields.ins().srcFieldVin);
            String startChargeTime = line.getString(ChargePredictFields.ins().srcFieldStartChargeTime);
            String chargeId = vin.concat(startChargeTime);
            Float bmsSoc = line.getFloat(ChargePredictFields.ins().srcFieldBmsSoc);
            String tidVal = line.getString(ChargePredictFields.ins().srcFieldTid);
            String s = timeOfLatest5State.get(chargeId);
            if (StringUtils.isBlank(s)){
                timeOfLatest5State.put(chargeId,tidVal);
                return;
            }
            //判断有几个点,不能大于5：判断现在有几个，等于5 和小于5
            String[] split = s.split(ChargePredictFields.ins().char3);
            if (split.length< DwsStrConstant.ins().num5){
                //小于5 拼接
                String concat = s.concat(ChargePredictFields.ins().char3).concat(tidVal);
                timeOfLatest5State.put(chargeId,concat);
                return;
            }
            //等于5 说明需要去掉开头的
            StringJoiner sj = new StringJoiner(ChargePredictFields.ins().char3);
            for (int i = 1; i <split.length ; i++) {
                String item = split[i];
                sj = sj.add(item);
            }
            sj.add(tidVal);
            timeOfLatest5State.put(chargeId,sj.toString());
        }catch (Exception e){
            e.printStackTrace();
        }

    }

    private void state2Line(JSONObject line) {
        try {
            String vin = line.getString(ChargePredictFields.ins().srcFieldVin);
            String startChargeTime = line.getString(ChargePredictFields.ins().srcFieldStartChargeTime);
            String chargeId = vin.concat(startChargeTime);
            //添加字段--predict_time字段   原本格式= 2022-09-10 17:05:22  临时用它存储每次预测的过程
            String keyPredict = ChargePredictFields.ins().stdFieldPredictTime;
            String stateKeyPredict = chargeId.concat(keyPredict);
            String predictTime=latestEstimateState.get(stateKeyPredict);
            line.put(keyPredict, predictTime);

            //添加字段-- predict_time_multidimension
            String keyMult = ChargePredictFields.ins().stdFieldPredictTimeV2;
            String stateKeyMult = chargeId.concat(keyMult);
            String predict= latestEstimateState.get(stateKeyMult);
            line.put(keyMult, predict);
        }catch (Exception e){
            e.printStackTrace();
        }
    }

    /**
     * 计算是否属于间隔点 soc每隔1%计算一次
     * @param line
     * @return
     */
    private boolean isNeededMatch(JSONObject line,float num) {
        try {
            //当前工况信息
            String vin = line.getString(ChargePredictFields.ins().srcFieldVin);
            String startChargeTime = line.getString(ChargePredictFields.ins().srcFieldStartChargeTime);
            Float bmsSoc = line.getFloat(ChargePredictFields.ins().srcFieldBmsSoc);
            String chargeId = vin.concat(startChargeTime);
            //判断状态
            String stateSoc = latestSocIntervalPointState.get(chargeId);
            if (StringUtils.isBlank(stateSoc)){
                //说明是第一次判断
                latestSocIntervalPointState.put(chargeId,String.valueOf(bmsSoc));
                line.put("start_point_flag",DwsStrConstant.ins().str1);
                line.put("soc_interval_point",DwsStrConstant.ins().str1);
                return true;
            }
            //判断是否符合间隔点信息
            float v = Float.parseFloat(stateSoc);
            float deltaV =bmsSoc - v ;
            if (deltaV>= num){
                //是间隔点
                latestSocIntervalPointState.put(chargeId,String.valueOf(bmsSoc));
                line.put("soc_interval_point",DwsStrConstant.ins().str1);
                return true;
            }
        }catch (Exception e){
            e.printStackTrace();
        }
        line.put("soc_interval_point",DwsStrConstant.ins().str0);
        return false;
    }

    /**
     * 计算初始或者更新
     * @param line
     * @return true -是更新阶段
     */
    private boolean isUpdatePhase(JSONObject line) {
        try {
            String vin = line.getString(ChargePredictFields.ins().srcFieldVin);
            Long tidVal = line.getLong(ChargePredictFields.ins().srcFieldTid);
            String startChargeTime = line.getString(ChargePredictFields.ins().srcFieldStartChargeTime);
            String bmsSoc = line.getString(ChargePredictFields.ins().srcFieldBmsSoc);
            String chargeId = vin.concat(startChargeTime);
            //如果true 说明可能是更新
            String socState = startSocState.get(chargeId);
            float v = Float.parseFloat(socState);
            float s = Float.parseFloat(bmsSoc);
            String obc = line.getString(ChargePredictFields.ins().srcFieldBmsObcConnect);
            int cnt = 5;
            if (StringUtils.equalsIgnoreCase(obc,DwsStrConstant.ins().str1)){
                //慢充的初始阶段设置为3
                cnt = 1;
            }
            float delta = s - v;

            //记录电流基础值
            boolean w = delta <= 2.1 && delta > 2.0;
            if (w){
                baseVal(line);
            }
            //区分初始计算和更新计算
            boolean b = delta < cnt;
            if (b){
                //与第一次记录的soc值 相差 小于等于5 即说明还是在初始预测阶段
                return false;
            }
            return true;
        }catch (Exception e){
            e.printStackTrace();
        }
        return false;
    }

    private void baseVal(JSONObject line) {
        try {
            String chargeIdName = chargeIdName(line);
            String s = timeOfLatest5State.get(chargeIdName);
            if (StringUtils.isBlank(s)){
                return;
            }
            String[] split = s.split(ChargePredictFields.ins().char3);
            String a = split[split.length - 1];
            String curr = a.split(ChargePredictFields.ins().colon)[1];
            String key = chargeIdName.concat("base_curr");
            timeOfLatest5State.put(key ,curr);
        }catch (Exception e){
            e.printStackTrace();
        }
    }

    /**
     *  判断是否大于1min
     * @param line
     * @return
     */
    private boolean oneMinute(JSONObject line) {
        try {
            String chargeIdName = chargeIdName(line);
            String vin = line.getString(ChargePredictFields.ins().srcFieldVin);
            String tid = line.getString(ChargePredictFields.ins().srcFieldTid);
            String curr = line.getString(ChargePredictFields.ins().srcFieldBmsBatterCurrRm);
            //电流稳定后的全新soc
            //P0 电流稳定
            //稳定开始 稳定结束
            String currSteadyStart = chargeIdName.concat("currsteadystart");
            String steadyEnd = chargeIdName.concat("currsteadyend");
            if (!startSocState.contains(currSteadyStart)){
                startSocState.put(currSteadyStart,tid.concat(ChargePredictFields.ins().char3).concat(curr));
                return false;
            }
            String[] sp1 = startSocState.get(currSteadyStart).split(ChargePredictFields.ins().char3);
            String tidLag1 = sp1[0];
            String currLag1 = sp1[1];
            float delta = Math.abs(Math.abs(Float.parseFloat(curr)) - Math.abs(Float.parseFloat(currLag1)));
            if (delta>0.5 && !startSocState.contains(steadyEnd)){
                //电流波动
                startSocState.put(currSteadyStart,tid.concat(ChargePredictFields.ins().char3).concat(curr));
                return false;
            }
            //电流稳定
            long tidDelta = Long.parseLong(tid) - Long.parseLong(tidLag1);
            if (tidDelta <60000 ){
                //不足一分钟后
                return false;
            }
            //一分钟后的第一个点 = 稳定结束
            String soc = line.getString(ChargePredictFields.ins().srcFieldBmsSoc);
            if (!startSocState.contains(steadyEnd)){
                startSocState.put(steadyEnd,tid.concat(ChargePredictFields.ins().char3)
                        .concat(curr).concat(ChargePredictFields.ins().char3)
                        .concat(soc));
                return false;
            }
            //P2 全新soc 0.1
            String[] sp2 = startSocState.get(steadyEnd).split(ChargePredictFields.ins().char3);
            String steadyEndSoc = sp2[2];
            float socDelta = Float.parseFloat(soc) - Float.parseFloat(steadyEndSoc);
            if (socDelta <0.09){
                return false;
            }
            //全新的第一个点作为起始soc
            if (!startSocState.contains(chargeIdName)){
                startSocState.put(chargeIdName,soc);
            }
            return true;
        }catch (Exception e){
            e.printStackTrace();
        }
        return false;
    }
    private boolean oneMinuteSoc01(JSONObject line) {
        try {
            String chargeId = chargeIdName(line);
            String format = DateFormatUtils.format(System.currentTimeMillis(), ChargePredictFields.ins().yyyyMmddss1);
            line.put(ChargePredictFields.ins().handleTime,format);
            //需求改为 soc +0.1的第一个点
            //需求又改为 一分钟  还说 我开发没有和文档保持一致
            String vin = line.getString(ChargePredictFields.ins().srcFieldVin);
            Long tid = line.getLong(ChargePredictFields.ins().srcFieldTid);
            Long startChargeTime = line.getLong(ChargePredictFields.ins().srcFieldStartChargeTime);
            String soc = line.getString(ChargePredictFields.ins().srcFieldBmsSoc);
            long l = tid - startChargeTime;
            boolean b = l >= 60000;
            if (b){
                //第一个点 是 一分钟之后
                if (!startSocState.contains(chargeId)){
                    startSocState.put(chargeId,soc);
                }
                return true;
            }
            //说明不足一分钟
            return false;
        }catch (Exception e){
            e.printStackTrace();
        }
        return false;
    }
    private boolean oneMinuteSoc(JSONObject line) {
        try {
            String chargeId = chargeIdName(line);
            String format = DateFormatUtils.format(System.currentTimeMillis(), ChargePredictFields.ins().yyyyMmddss1);
            line.put(ChargePredictFields.ins().handleTime,format);
            //需求改为 soc +0.1的第一个点
            //需求又改为 一分钟  还说 我开发没有和文档保持一致
            String vin = line.getString(ChargePredictFields.ins().srcFieldVin);
            Long tid = line.getLong(ChargePredictFields.ins().srcFieldTid);
            Long startChargeTime = line.getLong(ChargePredictFields.ins().srcFieldStartChargeTime);
            long l = tid - startChargeTime;
            String timeKey = chargeId.concat("oneMinute-time");
            boolean contains = startSocState.contains(timeKey);
            boolean timeB = false;
            if (!contains){
                boolean b = l >= 60000;
                if (b){
                    //第一个点 是 一分钟之后
                    startSocState.put(timeKey,tid.toString());
                    timeB = true;
                }
            }
            //判断soc
            Float bmsSoc = line.getFloat(ChargePredictFields.ins().srcFieldBmsSoc);
//            String stateKey = chargeId.concat("oneMinute")
            String stateKey = chargeId;
            String s = startSocState.get(stateKey);
            if (StringUtils.isBlank(s)){
                startSocState.put(stateKey,bmsSoc.toString());
                return false;
            }
            float sf = Float.parseFloat(s);
            float v = bmsSoc - sf;
            float i = 0.09f;
            if (v>i || timeB){
//                startSocState.put(stateKey,bmsSoc.toString())
                return true;
            }
        }catch (Exception e){
            e.printStackTrace();
        }
        return false;
    }

    private MapStateDescriptor<String, Map<String, Object>> confInfoMsd = new MapStateDescriptor<>("conf-info", BasicTypeInfo.STRING_TYPE_INFO, new MapTypeInfo<>(String.class, Object.class));
    private CloudMatch cloudMatch;
    private FirstMatch firstMatch;
    private MapState<String, String> initState;
    private MapState<String, String> latestEstimateState;
    private MapState<String, String> lastOneSliceState;
    private MapState<String, String> latestSocIntervalPointState;
    private MapState<String, String> startSocState;
    private MapState<String, String> timeOfLatest5State;
    @Override
    public void open(Configuration parameters) throws Exception {
        cloudMatch = new CloudMatch();
        firstMatch = new FirstMatch();
        initState = getRuntimeContext().getMapState(new MapStateDescriptor<>("initState", String.class, String.class));
        latestEstimateState = getRuntimeContext().getMapState(new MapStateDescriptor<>("latestPredictState", String.class, String.class));
        lastOneSliceState = getRuntimeContext().getMapState(new MapStateDescriptor<>("lastOneSliceState", String.class, String.class));
        latestSocIntervalPointState = getRuntimeContext().getMapState(new MapStateDescriptor<>("latestSocIntervalPointState", String.class, String.class));
        startSocState = getRuntimeContext().getMapState(new MapStateDescriptor<>("startSocState", String.class, String.class));
        timeOfLatest5State = getRuntimeContext().getMapState(new MapStateDescriptor<>("timeOfLatest5State", String.class, String.class));
    }



}
